package com.md.common.utils.redis;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.CollectionSerializer;
import com.esotericsoftware.kryo.serializers.JavaSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.time.*;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;

/**
 * 功能描述：redis使用kryo序列化
 *
 * @author lujianping@oak168.com
 * @version 1.0
 * @date 2018-6-15 11:51
 * @since JDK 1.8
 */
public class KryoSerialize {
    static Logger logger = LoggerFactory.getLogger(KryoSerialize.class);

    public static byte[] serialize(Object obj) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        try {
            Kryo kryo = new Kryo();
            registJava8Time(kryo);
            Output output = new Output(bos);
            kryo.writeObject(output, obj);
            byte[] bs = output.toBytes();
            output.flush();
            return bs;
        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            if (bos != null) {
                try {
                    bos.flush();
                    bos.close();
                } catch (IOException e) {
                    logger.error(e.getMessage());
                }
            }
        }
        return null;
    }

    private static void registJava8Time(Kryo kryo) {
        kryo.register(Duration.class);
        kryo.register(Instant.class);
        kryo.register(LocalDate.class);
        kryo.register(LocalTime.class);
        kryo.register(LocalDateTime.class);
        kryo.register(ZoneOffset.class);
        kryo.register(ZoneId.systemDefault().getClass()); // ZoneRegion
        kryo.register(ZoneId.class);
        kryo.register(OffsetTime.class);
        kryo.register(OffsetDateTime.class);
        kryo.register(ZonedDateTime.class);
        kryo.register(Year.class);
        kryo.register(YearMonth.class);
        kryo.register(MonthDay.class);
        kryo.register(Period.class);
    }

    public static <T> T deserialize(byte[] data, Class<T> clazz) {
        Kryo kryo = new Kryo();
        registJava8Time(kryo);
        ByteArrayInputStream bin = new ByteArrayInputStream(data);
        Input input = new Input(bin);
        T t = kryo.readObject(input, clazz);
        input.close();
        return t;
    }

    /**
     * 序列化列表
     *
     * @param obj   待序列化的列表
     * @param clazz 列表元素的类型
     * @param <T>   序列化的结果
     * @return
     */
    public static <T extends Serializable> byte[] serializationList(List<T> obj, Class<T> clazz) {
        Kryo kryo = new Kryo();
        kryo.setReferences(false);
        kryo.setRegistrationRequired(true);

        CollectionSerializer serializer = new CollectionSerializer();
        serializer.setElementClass(clazz, new JavaSerializer());
        serializer.setElementsCanBeNull(false);

        kryo.register(clazz, new JavaSerializer());
        kryo.register(ArrayList.class, serializer);

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        Output output = new Output(baos);
        kryo.writeObject(output, obj);
        output.flush();
        output.close();

        byte[] b = baos.toByteArray();
        try {
            baos.flush();
            baos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return Base64.getEncoder().encode(b);
    }

    /**
     * 反序列化列表
     *
     * @param obj   待反序列化的字符串
     * @param clazz 列表元素的类型
     * @param <T>   反序列化后的列表结果
     * @return
     */
    @SuppressWarnings("unchecked")
    public static <T extends Serializable> List<T> deserializationList(byte[] obj, Class<T> clazz) {
        Kryo kryo = new Kryo();
        kryo.setReferences(false);
        kryo.setRegistrationRequired(true);

        CollectionSerializer serializer = new CollectionSerializer();
        serializer.setElementClass(clazz, new JavaSerializer());
        serializer.setElementsCanBeNull(false);

        kryo.register(clazz, new JavaSerializer());
        kryo.register(ArrayList.class, serializer);

        ByteArrayInputStream bais = new ByteArrayInputStream(obj);
        Input input = new Input(bais);
        return (List<T>) kryo.readObject(input, ArrayList.class, serializer);
    }
}
